/*
* 作者: wupengxin
* 创建时间: 2016-08-20 18:34:26
* 说明: 输出到elasticsearch
*/
package outputplug

import (
	"common"
	//. "fmt"
	"strings"
	//"log"
	"errors"
	"bytes"
	"net/http"
	"encoding/json"
	"io/ioutil"
	"strconv"
	//"time"
)

type EsWrite struct {
	Hosts []string  // es的IP地址
	Index string    // es的索引名称
	Type string //文档类型
	Workers int // 创建多个worker同时发送数据
	BulkSize int // bulk的大小 
}

//配置es信息
func configEsWrite(conf map[string]string, es *EsWrite)  error{
	// 读取配置文件 设置属性
	if v, ok := conf["hosts"]; ok{
		es.Hosts = strings.Split(v, ",")
	}else {
		es.Hosts = []string{"http://127.0.0.1:9200"}
	}
	// Index
	if v, ok:= conf["index"]; ok{
		es.Index = v
	}else {
		err := errors.New("读取配置文件错误")
		common.Debug(err)
		return  err
	}
	// type
	if v, ok:= conf["type"]; ok{
		es.Type = v
	}else {
		err := errors.New("读取配置文件错误")
		common.Debug(err)
		return err
	}
	// workers
	if v, ok:= conf["workers"]; ok{
		es.Workers, _= strconv.Atoi(v)
	}else {
		es.Workers = 1
	}
	//bulk
	if v, ok:= conf["bulk"]; ok{
		es.BulkSize, _= strconv.Atoi(v)
	}else {
		es.BulkSize = 500
	}
	return nil
}


// 创建一个es对象
func NewEsWrite(conf map[string]string) (*EsWrite, error) {
	es := EsWrite{}
	err := configEsWrite(conf, &es)
	return &es, err
}

func (es *EsWrite)InsertIndex(ch chan map[string]string) {
	type respjson struct {
		//Errors bool `json:"errors"`
		Status int `json:"status"`
	}

	buf := bytes.NewBuffer(make([]byte, 0))
	bufCnt := 0
	cnt := 0
	//sartTime := time.Now()
	for {
		//等待传递已经格式化好的内容 发送到es服务器
		// 使用bulk模式发送到es的数据格式
		/*
			{"index" : {}} \n  // 当API里面提供了index和type 此处只需要写为空即可
			{待发送的文档内容} \n
			{"index" : {}} \n
			{第二份文档内容}\n
			...
		*/
		buf.WriteString(`{ "index": {}}`)
		buf.WriteString("\n")
		recv, ok := <-ch
		if !ok{
			common.Debug(errors.New("通道被关闭, 停止向ES发送内容"))
			break
		}
		buf.WriteString(recv["contant"])
		buf.WriteString("\n")
		bufCnt++
		// 由于一次http请求比较浪费资源 es提供了bulk的模式
		if bufCnt < es.BulkSize{
			continue
		}
		cnt++
		//Printf("启动时间:%d \t 第%d个bulk的时间:%f\n", sartTime, cnt, time.Now().Sub(sartTime).Seconds())
		bufCnt = 0
		// 执行到这里 说明已经攒满一桶数据了 直接发给es 如果有多个hosts 需要给多台es主机发送数据
		for _, host := range es.Hosts{
				if 0 != strings.Index(host, "http"){
					host = "http://" + host
				}
				if 1 == strings.Count(host, ":"){
					host = host + ":9200"
				}
				url := host+"/"+es.Index+"/"+es.Type+"/"+"_bulk"
				resp, err := http.Post(url, "text/plan", buf)
				if err != nil{
					common.Debug(err)
					return
				}
				body, err := ioutil.ReadAll(resp.Body)
				resp.Body.Close()
				// 解析body
				contant := respjson{}
				if err = json.Unmarshal(body, &contant); err!=nil{
					common.Debug(err)
					break
				}else {
					if contant.Status == 400{
						common.Debug(errors.New("请求ES服务发生错误"))
					}
					common.Debug(errors.New("status:"+strconv.Itoa(contant.Status)))
				}
		}
	}
}


// 实现输出插件接口
func (es *EsWrite)PutOutStream(conf map[string]string, channels []chan map[string]string)  error{
	// 读取配置
	newes, e := NewEsWrite(conf)
	if e!=nil{
		common.Debug(e)
		return e
	}
	// 为多个通道创建协程
	//for _, ch := range channels{
	//	if ch == nil{
	//		return errors.New("chan is nil ")
	//	}
	//	go newes.InsertIndex(ch)
	//}
	// 2016-10-02 17:37:11 改为使用单通道进行内容传递
	for i := 0; i < es.Workers; i++{
		go newes.InsertIndex(channels[0])
	}

	return nil
}

// 实现解析配置接口
func (es *EsWrite)ConfigPlus(conf interface{}) (map[string]string, error)  {
	/*
	* {"hosts":
		"port":....
	  }
	*/
	//	Hosts []string  // es的IP地址
	//Index string    // es的索引名称
	//Type string //文档类型
	//Workers int // 创建多个worker同时发送数据
	//BulkSize int // bulk的大小
	dat := conf.(map[string]interface{})
	tmp := make(map[string]string)
	// 开始解析配置
	if _, ok := dat["hosts"]; !ok{
		return nil, errors.New("elasticsearch host must be conf")
	}else {
		tmp["hosts"] = dat["hosts"].(string)
	}
	if v, ok := dat["index"]; ok{
		tmp["index"] = v.(string)
	}
	if v, ok := dat["type"]; ok{
		tmp["type"] = v.(string)
	}
	if v, ok := dat["workers"]; ok{
		tmp["workers"] = v.(string)
	}
	if v, ok := dat["bulk"]; ok{
		tmp["bulk"] = v.(string)
	}
	println(tmp)
	return tmp, nil
}
